package com.atguigu.gamll.realtime.app.func;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gamll.realtime.common.GmallConfig;
import com.atguigu.gamll.realtime.utils.DimUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * @author Lei
 * @version 1.0
 * @Date: 2022/2/23 10:56
 * @Description: 将侧输出流中的数据写到phoenix表中
 */
public class DimSink extends RichSinkFunction<JSONObject> {

    private Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 注册驱动
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        // 建立连接
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        // 获取输出目的地表名
        String sinkTable = jsonObj.getString("sink_table");
        // 获取要输出的数据
        JSONObject dataJsonObj = jsonObj.getJSONObject("data");
        // 拼接 upsert 语句
        String upsert = genUpsertSql(sinkTable, dataJsonObj);
        System.out.println("插入数据的sql：" + upsert);
        // 获取数据库操作对象
        PreparedStatement ps = null;
        try {
            ps = conn.prepareStatement(upsert);
            // 执行SQL语句
            ps.execute();
            // Phoenix默认需要手动提交
            conn.commit();
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("插入时发生了异常");
        } finally {
            // 释放资源
            if (ps != null) {
                ps.close();
            }
        }

        if ("update".equals(jsonObj.getString("type"))) {
            DimUtil.delCached(sinkTable, dataJsonObj.getString("id"));
        }
    }

    // 拼接插入数据的sql  dataJson:{"tm_name":"lala","id","14"}
    private String genUpsertSql(String tableName, JSONObject dataJsonObj) {
        Set<String> keys = dataJsonObj.keySet();
        Collection<Object> values = dataJsonObj.values();

        String upsertSql = "upsert into " + GmallConfig.HBASE_SCHEMA
                + "." + tableName
                + "(" + StringUtils.join(keys, ",")
                + ") values ('" + StringUtils.join(values, "','") + "')";
        return upsertSql;
    }
}
